package cn.mjnxin.iot.connector.handle;

import cn.mjnxin.iot.base.domain.ChannelData;
import cn.mjnxin.iot.base.domain.SocketData;
import cn.mjnxin.iot.base.log.LogWorker;
import cn.mjnxin.iot.base.monitor.CountMonitor;
import cn.mjnxin.iot.base.util.MixUtil;
import cn.mjnxin.iot.connector.cache.HostRuntimeCache;
import cn.mjnxin.iot.connector.context.ConnectorContext;
import cn.mjnxin.iot.connector.context.IPeerContext;
import cn.mjnxin.iot.connector.exception.GateNotFoundException;
import cn.mjnxin.iot.connector.exception.MessageNotSupportException;
import cn.mjnxin.iot.connector.message.IMessage;
import cn.mjnxin.iot.connector.message.IMessageHandle;
import cn.mjnxin.iot.connector.message.IMessageReceiver;
import cn.mjnxin.iot.transfer.TransferExchange;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufUtil;
import io.netty.channel.Channel;

/**
 * ChannelData处理器
 * @author CLG 2023年10月10日
 * @version V1.0.0
 */
public class ChannelDataHandle implements IChannelDataHandle {
    private final IHandleFactory messageHandleFactory;
    private final IPeerContext context;
    private final boolean isHost;

    public ChannelDataHandle(IPeerContext context, IHandleFactory messageHandleFactory, boolean isHost) {
        this.messageHandleFactory = messageHandleFactory;
        this.context = context;
        this.isHost = isHost;
    }

    @Override
    public ChannelData resolve(ChannelData request) {
        try {
            context.setCurrentPeerAddress(request.getTerminalAddress());
            final SocketData socketData = request.getSocketData();
            final int pid = socketData.getPid();
    		final Integer logLevel = request.getLogLevel();

            ByteBuf byteBuf = socketData.getByteBuf();

            byte[] buf = new byte[byteBuf.readableBytes()];
            byteBuf.readBytes(buf);

            // 将requestBytes中的可读字节 转换成16进制数字符串
            String messageHex = ByteBufUtil.hexDump(buf).toUpperCase();
            if (LogWorker.isLogable()) {
            	LogWorker.log("接收报文:" + messageHex);
            }

            byte[] responseBytes = null;
            String responseHex = null;
            // 校验报文长度和起始字节是否有效
            IMessageHandle<? extends IMessage<?>> messageHandle = messageHandleFactory.getMessageHandle(pid);
            IMessage<?> message = null;
            if (messageHandle.validate(messageHex)) {
            	@SuppressWarnings("unchecked")
                final IMessageReceiver<IMessage<?>> messageCreator = (IMessageReceiver<IMessage<?>>) messageHandle.getMessageReceiver();
                // 解析应答报文业务数据得到对应的应答报文对象
                try {
                    message = messageCreator.onReceiveMessage(buf, messageHex);
                    if (message == null) {
                        //无应答或者buf本身就是应答报文则无需处理
                        return null;
                    }
                    responseBytes = message.pack();
                    responseHex =  message.getBodyHex();
                    //应答报文打印json格式内容
                    if (LogWorker.isLogable()) {
                    	LogWorker.log("应答报文:" + message.toJson());
                    }
                    //应答报文上报日志
                    messageHandle.logMessage(message, false);
                } 
                catch(MessageNotSupportException e) {
                    LogWorker.error(messageHex + "无法识别", e);
                    message = e.getTarget();
                    if (message == null) {
                    	//无应答或者buf本身就是应答报文则无需处理
                    	LogWorker.error(messageHex + " 处理出错，不返回", e);
                    	return null;
                    }
                    responseBytes = message.pack();
                    responseHex = MixUtil.bytes2hex(responseBytes);
                	LogWorker.logForce("异常应答:" + responseHex);
                }catch (Exception e) {
                    if (message == null) {
                        //无应答或者buf本身就是应答报文则无需处理
                        LogWorker.error(messageHex + " 处理出错，不返回", e);
                        return null;
                    }
                    
                    LogWorker.error("业务处理出错", e);
                    responseBytes = buf;
                    responseHex = messageHex;
                	LogWorker.logForce("原样返回:" + responseHex);
                }
            }
            if (responseBytes == null) {
                // 格式无效报文不处理，原样返回
                responseBytes = buf;
                responseHex = messageHex;
                LogWorker.logForce("无效报文");
            	LogWorker.logForce("原样返回:" + responseHex);
            }

            int pos = CountMonitor.hostRecieveCount.addAndGet(1);
            if (pos == 1) {
                // 初始化开始时间
                CountMonitor.hostRecieveStartTime = System.currentTimeMillis();
            }

            if (pos == CountMonitor.ThreadNum) {
                // 计算结束时间
                LogWorker.log(
                        "接收数据总花费时间为:" + (System.currentTimeMillis() - CountMonitor.hostRecieveStartTime) + "毫秒");
            }
            if (pos == 8000) {
                // 计算结束时间
                LogWorker.log(
                        "接收数据总花费时间为:" + (System.currentTimeMillis() - CountMonitor.hostRecieveStartTime) + "毫秒");
            }
            
            /**
             * 通过channel写出数据 会直接执行handler链中最后一个handler开始逆向执行
             */
            SocketData data = new SocketData(responseBytes);
            data.setPid(pid);
            data.setCount(socketData.getCount());

            if (message == null) {
                //无应答或者buf本身就是应答报文则无需处理
                LogWorker.error(messageHex + " 处理出错，不返回");
                return null;
            }
            
            //重新获取连接参数，如果报错就要判定断网
            String terminalKey = message.getConnectorId();
            String gateAddress = request.getGateAddress();
            String terminalAddress = request.getTerminalAddress();
            int time = 0;
            while(true) {
                try{
                	if (isHost) {
                		context.getGateAddress(terminalAddress);
                	}
                    break;
                }
                catch(GateNotFoundException ex) {
                    time++;
                    try {
                        Thread.sleep(1000L);
                    } catch (InterruptedException e1) {
                    }
                    if (time <= 6) {
                        terminalAddress = context.getPeerAddress(terminalKey);
                        continue;
                    }
                    LogWorker.log("检测到 peerAddress: " + terminalAddress + " 已经失效，断连此Peer");
                    final String currentTerminalAddress = terminalAddress;
                    new Thread() {
                        public void run() {
                            try {
                            	ByteBuf unlinkBuf = null;
                            	if (isHost) {
									unlinkBuf = TransferExchange.hostCloseTerminal(HostRuntimeCache.gateNo, pid, logLevel, currentTerminalAddress, gateAddress);
                            	}
                                Channel channel = HostRuntimeCache.getGateChannel(gateAddress);
                                if (channel == null) {
                                    LogWorker.error(ConnectorContext.CONNECTOR_NAME + "与网关断连，无法发送报文");
                                    return;
                                }
                            	if (isHost) {
	                                LogWorker.log("发送断连Client控制报文");
	                                channel.writeAndFlush(unlinkBuf);
	
	                                context.removePeer(currentTerminalAddress);
                            	}
                            } catch (Exception e) {
                                LogWorker.error(ex.getMessage(), ex);
                            }
                        }
                    }.start();
                    
                    return null;
                }
            }
//            int time = 0;
//            while(true) {
//                currentClientAddr = context.getClientAddress(clientKey);
//                if (currentClientAddr == null) {
//                    LogWorker.logForce(String.format("%s 检测到链路:%s 已经失效，等待重试", clientKey, clientAddress));
//                    time++;
//                }
//                else {
//                    break;
//                }
//
//                try {
//                    Thread.sleep(100L);
//                } catch (InterruptedException e1) {
//                }
//                if (time <= 6) {
//                    continue;
//                }
//                else {
//                    LogWorker.logForce(String.format("%s 检测到链路:%s 已经失效，判定断网", clientKey, clientAddress));
//                    new Thread() {
//                        public void run() {
//                            try {
//                                ByteBuf unlinkBuf = GateHostExchange.hostCloseClient(HostCacheQueue.gateNum, pid, clientAddress, gateAddress);
//                                
//                                Channel channel = HostCacheQueue.getGateChannel(gateAddress);
//                                if (channel == null) {
//                                    LogWorker.error(ConnectorContext.CONNECTOR_NAME + "与网关断连，无法发送报文");
//                                    throw new NetworkException("充电桩故障");
//                                }
//                                LogWorker.logForce("发送断连Client:"+clientKey+" 报文");
//                                channel.writeAndFlush(unlinkBuf);
//                                
//                            } catch (Exception e) {
//                                LogWorker.error(e.getMessage(), e);
//                            }
//                            context.delClient(clientAddress);
//                        }
//                    }.start();
//                    
//                    return null;
//                }
//            }
            
            return new ChannelData(gateAddress, terminalAddress, pid, request.getLogLevel(), data);
        } finally {
            context.setCurrentPeerAddress(null);
        }
    }
}
